#!/bin/bash

echo "=========================================="
echo "Kafka Topic和消费组设置脚本"
echo "=========================================="

# 检查Kafka服务是否运行
echo "检查Kafka服务状态..."
if ! nc -z localhost 9092 2>/dev/null; then
    echo "错误: Kafka服务未运行，请先启动Kafka服务"
    echo "启动命令:"
    echo "  zookeeper-server-start /usr/local/etc/kafka/zookeeper.properties &"
    echo "  kafka-server-start /usr/local/etc/kafka/server.properties &"
    exit 1
fi

echo "Kafka服务运行正常 ✓"

# 创建access-log topic
echo ""
echo "创建topic: access-log"
kafka-topics --create \
    --bootstrap-server localhost:9092 \
    --topic access-log \
    --partitions 3 \
    --replication-factor 1 \
    --if-not-exists

if [ $? -eq 0 ]; then
    echo "Topic创建成功 ✓"
else
    echo "Topic可能已存在，继续..."
fi

# 验证topic创建
echo ""
echo "验证topic信息..."
kafka-topics --describe \
    --bootstrap-server localhost:9092 \
    --topic access-log

# 显示消费组信息
echo ""
echo "=========================================="
echo "设置完成！"
echo "=========================================="
echo "Topic: access-log (3分区, 1副本)"
echo "消费组: group002 (将在首次消费时自动创建)"
echo ""
echo "测试命令:"
echo "1. 发送测试数据:"
echo "   kafka-console-producer --bootstrap-server localhost:9092 --topic access-log"
echo ""
echo "2. 消费数据:"
echo "   kafka-console-consumer --bootstrap-server localhost:9092 --topic access-log --group group002 --from-beginning"
echo ""
echo "3. 运行Flink程序:"
echo "   ./run-hot-items.sh"
echo "=========================================="
